package com.eshipenet.shipnet.equipmentextbuilder

import org.joda.time.DateTime
import play.api.libs.json._

import scala.util.{Failure, Success, Try}

object DebeziumParser {

  final case class UnknownOpError(result: JsLookupResult) extends Exception(result.toString)
  final case class InvalidValue(value: JsLookupResult)    extends Exception(value.toString)

  sealed trait Action {
    val timestamp: DateTime
  }
  case class Insert(after: JsObject, timestamp: DateTime)                   extends Action
  case class Update(before: JsObject, after: JsObject, timestamp: DateTime) extends Action
  case class Delete(before: JsObject, timestamp: DateTime)                  extends Action

  def apply(doc: JsValue): Try[Option[Action]] = doc \ "payload" \ "op" match {
    case JsDefined(JsString("c")) => parseInsert(doc).map(Some(_))
    case JsDefined(JsString("i")) => parseInsert(doc).map(Some(_))
    case JsDefined(JsString("r")) => parseInsert(doc).map(Some(_))
    case JsDefined(JsString("u")) => parseUpdate(doc).map(Some(_))
    case JsDefined(JsString("d")) => parseDelete(doc).map(Some(_))
    case JsUndefined()            => Success(None)
    case unknown                  => Failure(UnknownOpError(unknown))
  }

  private def parseInsert(doc: JsValue): Try[Insert] = {
    val schemas = parseTopLevelSchema(doc, "after")
    doc \ "payload" \ "after" match {
      case JsDefined(obj: JsObject) =>
        val timestamp = doc \ "payload" \ "ts_ms"
        Success(Insert(parseObject(obj, schemas), timestamp.as[DateTime]))
      case others => Failure(InvalidValue(others))
    }
  }

  private def parseUpdate(doc: JsValue): Try[Update] = {
    val beforeSchemas = parseTopLevelSchema(doc, "before")
    val afterSchemas  = parseTopLevelSchema(doc, "after")
    val before = doc \ "payload" \ "before" match {
      case JsDefined(obj: JsObject) =>
        parseObject(obj, beforeSchemas)
      case _ => Json.obj()
    }
    doc \ "payload" \ "after" match {
      case JsDefined(obj: JsObject) =>
        val after     = parseObject(obj, afterSchemas)
        val timestamp = doc \ "payload" \ "ts_ms"
        Success(Update(before, after, timestamp.as[DateTime]))
      case others => Failure(InvalidValue(others))
    }
  }

  private def parseDelete(doc: JsValue): Try[Delete] = {
    val beforeSchemas = parseTopLevelSchema(doc, "before")
    doc \ "payload" \ "before" match {
      case JsDefined(obj: JsObject) =>
        val timestamp = doc \ "payload" \ "ts_ms"
        Success(Delete(parseObject(obj, beforeSchemas), timestamp.as[DateTime]))
      case others => Failure(InvalidValue(others))
    }
  }

  private implicit val format: Format[FieldSchema] = Json.format[FieldSchema]

  private case class FieldSchema(field: String, name: Option[String], fields: Option[List[FieldSchema]])

  private def parseObject(obj: JsObject, schemas: List[FieldSchema]): JsObject = {
    val newObj = obj.fields map {
      case (key, str: JsString) => (key, parseJSON(key, str, schemas))
      case (key, value)         => (key, value)
    }
    JsObject(newObj)
  }

  private def parseJSON(key: String, str: JsString, schemas: List[FieldSchema]): JsValue = {
    schemas.find(_.field == key) match {
      case Some(s) =>
        if (s.name.nonEmpty && s.name.get.toLowerCase().contains("json")) {
          Try(Json.parse(str.value)) match {
            case Success(v) => v
            case _          => str
          }
        } else {
          str
        }
      case _ => str
    }
  }

  private def parseTopLevelSchema(doc: JsValue, field: String): List[FieldSchema] =
    doc \ "schema" \ "fields" match {
      case JsDefined(JsArray(fields)) =>
        fields
          .map(_.asOpt[FieldSchema])
          .find(s => s.nonEmpty && s.get.field == field)
          .flatMap {
            case Some(s) => s.fields
            case _       => None
          }
          .getOrElse(List())
      case _ => List()
    }
}
